package com.erp.service.impl;

import com.erp.service.RocketMQTemplateProducer;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.util.Map;

@Service
public class RocketMQTemplateProducerImpl implements RocketMQTemplateProducer {

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @Override
    public void send(String topicName, String data) {
        send(topicName, data, (Map<String, Object>) null);
    }

    @Override
    public void send(String topicName, String data, Map<String, Object> headers) {
        send(topicName, null, data, headers);
    }

    @Override
    public void send(String topicName, String tags, String data) {
        send(topicName, tags, data, null);
    }

    @Override
    public void send(String topicName, String tags, String data, Map<String, Object> headers) {
        if (StringUtils.isBlank(tags)) {
            rocketMQTemplate.convertAndSend(topicName, data, headers);
            return;
        }
        rocketMQTemplate.convertAndSend(topicName + ":" + tags, data, headers);
    }

    @Override
    public void send(String topicName, Message<?> message) {
        send(topicName, message, null);
    }

    @Override
    public void send(String topicName, Message<?> message, Map<String, Object> headers) {
        send(topicName, null, message, headers);
    }

    @Override
    public void send(String topicName, String tags, Message<?> message) {
        send(topicName, tags, message, null);
    }

    @Override
    public void send(String topicName, String tags, Message<?> message, Map<String, Object> headers) {
        if (!MapUtils.isEmpty(headers)) {
            headers.forEach(message.getHeaders()::put);
        }
        if (StringUtils.isBlank(tags)) {
            rocketMQTemplate.syncSend(topicName, message);
        } else {
            rocketMQTemplate.syncSend(topicName + ":" + tags, message);
        }
    }

    @Override
    public void send(String topicName, Message<?> message, int delayLevel) {
        send(topicName, message, null, delayLevel);
    }

    @Override
    public void send(String topicName, Message<?> message, Map<String, Object> headers, int delayLevel) {
        send(topicName, null, message, headers, delayLevel);
    }

    @Override
    public void send(String topicName, String tags, Message<?> message, int delayLevel) {
        send(topicName, tags, message, null, delayLevel);
    }

    @Override
    public void send(String topicName, String tags, Message<?> message, Map<String, Object> headers, int delayLevel) {
        if (!MapUtils.isEmpty(headers)) {
            headers.forEach(message.getHeaders()::put);
        }
        if (StringUtils.isBlank(tags)) {
            rocketMQTemplate.syncSend(topicName, message, rocketMQTemplate.getProducer().getSendMsgTimeout(), delayLevel);
        } else {
            rocketMQTemplate.syncSend(topicName + ":" + tags, message, rocketMQTemplate.getProducer().getSendMsgTimeout(), delayLevel);
        }
    }

    @Override
    public void send(String topicName, String tags, Message<?> message, String hashKey, Map<String, Object> headers, int delayLevel) {
        if (!MapUtils.isEmpty(headers)) {
            headers.forEach(message.getHeaders()::put);
        }
        String destination = StringUtils.isBlank(tags) ? topicName : topicName + ":" + tags;
        int timeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        rocketMQTemplate.syncSendOrderly(destination, message, hashKey, timeout, delayLevel);
    }
}
